<?php

namespace ali\mq;

use app\api\mq\handler\HandlerInterface;
use app\api\mq\struct\BaseStruct;
use MQ\Exception\AckMessageException;
use MQ\Exception\MessageNotExistException;
use MQ\Model\Message;
use MQ\Model\TopicMessage;
use MQ\MQClient;
use think\Exception;
use think\facade\Log;

/**
 * 消费者没有实现幂等
 */
class RocketMq
{
    private $endPoint;
    private $accessId;
    private $accessKey;
    private $instanceId;
    private $groupId;

    public function __construct()
    {
        $config = config('ali_qmh');
        $this->endPoint = $config['mq']['endPoint'];
        $this->accessId = $config['mq']['accessKeyId'];
        $this->accessKey = $config['mq']['accessKeySecret'];
        $this->instanceId = $config['mq']['instanceId'];
        $this->groupId = $config['mq']['gid'];
    }

    /**
     * @param BaseStruct $message
     * @return TopicMessage|bool
     */
    public static function send($topic, BaseStruct $message) {
        try {
            $that = new self();
            $client = new MQClient($that->endPoint, $that->accessId, $that->accessKey);

            $instanceId = $that->instanceId;
            $producer = $client->getProducer($instanceId, $topic);
            $publishMessage = new TopicMessage($message->getJson());

            // 设置消息的自定义属性。
            // $publishMessage->putProperty("a", $i);
            // 设置消息的Key。
            // $publishMessage->setMessageKey("MessageKey");

            $result = $producer->publishMessage($publishMessage);
            return $result;
        } catch (\Exception $e) {
            Log::error('消息发送失败');
            Log::error($message->getJson());
            Log::error($e->getMessage());
            return false;
        }
    }

    public static function customer($topic) {
        $that = new self();
        $client = new MQClient($that->endPoint, $that->accessId, $that->accessKey);
        $consumer = $client->getConsumer($that->instanceId, $topic, $that->groupId);

        $config = config('ali_qmh.mq');
        /* @var HandlerInterface $handlerClass */
        $handlerClass = $config['topic'][$topic]['handler'];

        $structClass = $config['topic'][$topic]['struct'];

        // 在当前线程循环消费消息，建议多开个几个线程并发消费消息。
        while (True) {
            try {
                // 长轮询消费消息。
                // 长轮询表示如果Topic没有消息,则请求会在服务端挂起3s，3s内如果有消息可以消费则立即返回客户端。
                $messages = $consumer->consumeMessage(
                    3, // 一次最多消费3条（最多可设置为16条）。
                    3  // 长轮询时间3秒（最多可设置为30秒）。
                );
            } catch (\Exception $e) {
                if ($e instanceof MessageNotExistException) {
                    // Topic中没有消息可消费，继续轮询。
                    continue;
                }

                Log::error('解析接收失败');
                Log::error($e->getMessage());

                sleep(3);
                continue;
            }

            // 处理业务逻辑。
            $receiptHandles = array();
            foreach ($messages as $message) {
                try {
                    /* @var Message $message */
                    $body = $message->getMessageBody();
                    $body = json_decode($body, true) ?? [];

                    $structClass = $structClass::create($body);
                    $res = $handlerClass::handler($structClass);
                    if (!$res) {
                        throw new Exception(\Debug::getDebug());
                        //continue;
                    }
                } catch (\Exception $e) {
                    Log::error('消息队列 handler 处理失败:' . $message->getMessageId() . json_encode($message->getMessageBody()));
                    Log::error($e->getMessage() . ' in '. $e->getFile() . ': ' . $e->getLine());
                }
                $receiptHandles[] = $message->getReceiptHandle();
            }

            try {
                $consumer->ackMessage($receiptHandles);
            } catch (\Exception $e) {
                if ($e instanceof AckMessageException) {
                    // 某些消息的句柄可能超时，会导致消息消费状态确认不成功。
                    Log::error(sprintf("Ack Error, RequestId:%s\n", $e->getRequestId()));
                    foreach ($e->getAckMessageErrorItems() as $errorItem) {
                        Log::error(sprintf("\tReceiptHandle:%s, ErrorCode:%s, ErrorMsg:%s\n", $errorItem->getReceiptHandle(), $errorItem->getErrorCode(), $errorItem->getErrorCode()));
                    }
                }
            }
        }
    }
}